package com.hw.nio.netty.rpc;


import com.alibaba.dubbo.common.io.Bytes;
import com.hw.nio.netty.ByteBufUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageCodec;

import java.io.*;
import java.util.List;

/**
 * RPC 编码解码
 */
public class RpcCodec extends ByteToMessageCodec<Object> {

    protected static final int HEADER_LENGTH = 16;
    // 协议标识
    protected static final short MAGIC = 0xdad;
    protected static final ByteBuf MAGIC_BUF = Unpooled.copyShort(MAGIC);
    protected static byte FLAG_REQUEST = (byte) 0x80;//1000 0000
    protected static final byte FLAG_TWO_WAY = (byte) 0x40; //0100 0000
    protected static final byte FLAG_EVENT = (byte) 0x20;  //0010 0000
    protected static final int SERIALIZATION_MASK = 0x1f;  //0001 1111


    @Override
    protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
        if (msg instanceof Transfer) {
            doEncode((Transfer) msg, out);
        }
    }


    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        Transfer transfer = doDecode(in);
        if (transfer != null) {
            out.add(transfer);
        }
    }

    public Transfer doDecode(ByteBuf in) {
        int index = ByteBufUtil.indexOf(in,MAGIC_BUF);
        if (index < 0) {
            // 需要更多字节
            return null;
        }

        if (!in.isReadable(index + HEADER_LENGTH)) {
            return null;
        }

        byte[] header = new byte[HEADER_LENGTH];
        ByteBuf slice = in.slice();
        slice.readBytes(header);
        int length = Bytes.bytes2int(header, 12);
        int allLength = index + HEADER_LENGTH + length;
        if (!in.isReadable(allLength)) {
            return null;
        }
        Transfer transfer = new Transfer(Bytes.bytes2long(header,4));
        transfer.heartbeat = (header[2] & FLAG_EVENT) != 0;
        transfer.request = (header[2] & FLAG_REQUEST) != 0;
        transfer.twoWay = (header[2] & FLAG_TWO_WAY) != 0;
        // 通过 & 运算得出字节的数值
        transfer.serializableId = (byte) (header[2] & SERIALIZATION_MASK);
        transfer.status = header[3];
        if (!transfer.heartbeat) {
            byte content[] = new byte[length];
            slice.readBytes(content);
            transfer.target = deserialize(transfer.serializableId, content);
        }
        in.skipBytes(allLength);
        return transfer;
    }

    private Object deserialize(byte serializableId, byte[] content) {
        if (serializableId == Transfer.SERIALIZABLE_JAVA) {
            try {
                ObjectInputStream objectInputStream = new ObjectInputStream(new ByteArrayInputStream(content));
                return objectInputStream.readObject();
            } catch (IOException | ClassNotFoundException e) {
                e.printStackTrace();
                throw new UnsupportedOperationException();
            }
        } else {
            throw new UnsupportedOperationException();
        }
    }


    public void doEncode(Transfer msg, ByteBuf out) {
        byte[] header = new byte[HEADER_LENGTH];
        Bytes.short2bytes(MAGIC, header);
        header[2] = msg.serializableId;
        if (msg.request) header[2] |= FLAG_REQUEST;
        if (msg.twoWay) header[2] |= FLAG_TWO_WAY;
        if (msg.heartbeat) header[2] |= FLAG_EVENT;
        if (!msg.request) header[3] = msg.status;

        Bytes.long2bytes(msg.id, header, 4);
        int len = 0;
        byte[] body = new byte[0];
        if (!msg.heartbeat) {
            body = serialize(msg.serializableId, msg.target);
            len = body.length;
        }

        Bytes.int2bytes(len, header, 12);
        out.writeBytes(header);
        out.writeBytes(body);
    }


    private byte[] serialize(byte serializableId, Object target) {
        if (serializableId == Transfer.SERIALIZABLE_JAVA) {
            ByteArrayOutputStream bout = null;
            try {
                bout = new ByteArrayOutputStream();
                ObjectOutputStream objectOutputStream = new ObjectOutputStream(bout);
                objectOutputStream.writeObject(target);
            } catch (Exception e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
            return bout.toByteArray();
        }

        return new byte[0];
    }

    public static void main(String[] args) {
        byte[] header = new byte[HEADER_LENGTH];
        System.out.println(FLAG_REQUEST |= FLAG_REQUEST);
    }
}
